package com.duobaoyu.dby.supply.job.task.supply;

import com.alibaba.fastjson.JSON;
import com.duobaoyu.dby.channel.adapter.enums.DistributionQueueJobStatusEnum;
import com.duobaoyu.dby.channel.adapter.enums.biz.message.MessageStatusEnum;
import com.duobaoyu.dby.common.core.constant.CommonConstants;
import com.duobaoyu.dby.common.core.constant.StringPool;
import com.duobaoyu.dby.common.mybatis.util.EnumUtils;
import com.duobaoyu.dby.supply.biz.converter.ChannelMessageConverter;
import com.duobaoyu.dby.supply.biz.service.message.MessageService;
import com.duobaoyu.dby.supply.biz.service.queue.core.DistributionQueueJobConsumerWork;
import com.duobaoyu.dby.supply.biz.strategy.AbstractChannelMessageAsyncHandler;
import com.duobaoyu.dby.supply.biz.util.DingDingExtUtils;
import com.duobaoyu.dby.supply.common.biz.bo.message.ChannelMessageBO;
import com.duobaoyu.dby.supply.core.common.exception.CustomAssert;
import com.duobaoyu.dby.supply.core.framework.util.FrameworkUtils;
import com.duobaoyu.dby.supply.core.rocketmq.constant.RocketMqConstant;
import com.duobaoyu.dby.supply.db.entity.message.ChannelMessageOrder;
import com.duobaoyu.dby.supply.db.entity.message.DelayMessage;
import com.duobaoyu.dby.supply.db.entity.message.MqMessage;
import com.duobaoyu.dby.supply.db.repository.message.ChannelMessageRepository;
import com.duobaoyu.dby.supply.db.repository.message.DelayMessageRepository;
import com.google.common.collect.Lists;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Arrays;
import java.util.List;

/**
 * @author liyang
 * @version 1.0.0
 * @ClassName SupplyChannelMessageRetryTask
 * @Description 供应链消息兜底任务
 * @createTime 17:10:00 2023/10/26
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class SupplyChannelMessageRetryTask{

    private final MessageService messageService;
    private final DelayMessageRepository delayMessageRepository;
    private final ChannelMessageConverter channelMessageConverter;
    private final ChannelMessageRepository channelMessageRepository;
    private final DistributionQueueJobConsumerWork distributionQueueJobConsumerWork;

    /**
     * Rocket MQ 发送失败重试
     */
    @XxlJob("sendRetryMessageJobHandler")
    public void sendRetryMessageJobHandler() {
        String tag = XxlJobHelper.getJobParam();
        List<String> tags = null;
        if(StringUtils.isNotBlank(tag)) {
            tags = Arrays.asList(tag.split(StringPool.COMMA));
        }
        int skip = CommonConstants.ZERO;
        int limit = CommonConstants.HUNDRED;
        while (true) {
            List<MqMessage> messageList = messageService.mqMessageList(skip,limit,Lists.newArrayList(RocketMqConstant.FAIL, RocketMqConstant.SENDING),tags);
            if(CollectionUtils.isEmpty(messageList)) {
                break;
            }
            LocalDateTime dateTime = LocalDateTime.now().minusMinutes(CommonConstants.FOUR);
            for (MqMessage message : messageList) {
                if(message.getCreateTime().isBefore(dateTime)) {
                    boolean result = false;
                    try {
                        result = messageService.retryRocketMqMessage(message.getId(),message);
                    } catch (Exception e) {
                        log.warn("rocketmq消息重试异常: messageId: {},. error => {}",message.getId(), ExceptionUtils.getStackTrace(e));
                    }
                    if(!result) {
                        log.warn("rocketmq消息重试发送失败: message id = {} retry send failed",message.getId());
                    }
                }
            }
            skip += limit;
        }
    }

    /**
     * 渠道消息兜底处理任务
     */
    @XxlJob("channelMessageJobHandler")
    public void channelMessageJobHandler() {
        int skip = CommonConstants.ZERO;
        int limit = CommonConstants.HUNDRED;
        while (true) {
            List<ChannelMessageOrder> channelMessageOrders = channelMessageRepository.channelMessageOrderList(skip,limit, Lists.newArrayList(MessageStatusEnum.WAIT_HANDLE.getValue(), MessageStatusEnum.FAIL.getValue()), CommonConstants.FOUR);
            if(CollectionUtils.isEmpty(channelMessageOrders)) {
                break;
            }
            for (ChannelMessageOrder message : channelMessageOrders) {
                if(EnumUtils.eq(MessageStatusEnum.FAIL,message.getStatus()) && message.getRetryCount() > CommonConstants.TWO) {
                    DingDingExtUtils.sendChannelMessageRetry(message);
                    continue;
                }
                boolean result = false;
                try {
                    ChannelMessageBO bo = channelMessageConverter.toChannelMessage(message);
                    AbstractChannelMessageAsyncHandler handler = FrameworkUtils.strategyInstance(AbstractChannelMessageAsyncHandler.class, bo.getChannelCode(), bo.getMessageType());
                    CustomAssert.notNull(handler, String.format("渠道消息异步处理器[AbstractChannelMessageAsyncHandler]不存在(channelCode:%s , messageType:%s)", bo.getChannelCode(), bo.getMessageType()));
                    result = handler.strategyHandler(bo);
                } catch (Exception e) {
                    log.warn("渠道消息兜底处理任务处理异常: messageId:{} , error => {}",message.getId(),ExceptionUtils.getStackTrace(e));
                }
                if(!result) {
                    log.warn("渠道消息处理失败 message: {}", JSON.toJSONString(message));
                }
            }
            skip += limit;
        }
    }

    /**
     * 供应链延时消息兜底处理任务
     */
    @XxlJob("supplyDelayMesageJobHandler")
    public void supplyDelayMesageJobHandler() {
        int skip = CommonConstants.ZERO;
        int limit = CommonConstants.HUNDRED;
        while (true) {
            long currentMilli = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli();
            List<DelayMessage> delayMessages = delayMessageRepository.delayMessageList(skip, limit, Lists.newArrayList(DistributionQueueJobStatusEnum.WAIT_HANDLE.getValue(), DistributionQueueJobStatusEnum.FAILURE.getValue()),currentMilli,CommonConstants.FOUR);
            if(CollectionUtils.isEmpty(delayMessages)) {
                break;
            }
            if(CollectionUtils.isNotEmpty(delayMessages)) {
                for (DelayMessage delayMessage : delayMessages) {
                    try {
                        distributionQueueJobConsumerWork.consume(delayMessage);
                    } catch (Exception e) {
                        log.warn("供应链延时消息兜底处理任务: messageId:{} , error => {}",delayMessage.getId(),ExceptionUtils.getStackTrace(e));
                    }
                }
            }
            skip += limit;
        }
    }
}
